package com.atguigu.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.bean.TableProcessDim;
import com.atguigu.edu.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>> {
    //将配置表中的配置信息预加载到程序中，解决主流数据先到，广播数据后到的问题
    private Map<String,TableProcessDim> configMap = new HashMap();

    private MapStateDescriptor<String, TableProcessDim> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcessDim> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Connection connection = JdbcUtil.getMySqlConnection();
        String sql = "select * from edu_config.table_process_dim";
        List<TableProcessDim> tableProcessDimList = JdbcUtil.queryList(connection, sql, TableProcessDim.class, true);
        for (TableProcessDim tableProcessDim : tableProcessDimList) {
            configMap.put(tableProcessDim.getSourceTable(),tableProcessDim);
        }
        JdbcUtil.closeMySqlConnection(connection);
    }

    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {
        //获取业务数据库维表的表名
        String table = value.getString("table");
        System.out.println(table);
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据表名到广播状态中获取对应的配置信息
        TableProcessDim tableProcessDim = null;
        if ((tableProcessDim = broadcastState.get(table)) != null
                || (tableProcessDim = configMap.get(table)) != null) {
            JSONObject dataJsonObj = value.getJSONObject("data");
            //在向下游传递数据之前，过滤掉不需要传递的字段
            String sinkColumns = tableProcessDim.getSinkColumns();
            deleteColNotNeed(dataJsonObj,sinkColumns);
            //在向下游传递数据之前，补充对业务数据库维度表进行的操作类型
            String type = value.getString("type");
            dataJsonObj.put("type", type);
            //向下游写出数据
            Tuple2<JSONObject, TableProcessDim> record = new Tuple2<>(dataJsonObj, tableProcessDim);
            out.collect(record);
        }
    }

    //处理广播流数据 将配置信息放到广播状态中去或者从广播状态中删除对应的配置
    @Override
    public void processBroadcastElement(TableProcessDim value, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {
        //获取广播状态
        BroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //获取配置信息中的op操作类型
        String op = value.getOp();
        String sourceTableName = value.getSourceTable();
        //根据op确定对广播状态作何种操作
        if ("d".equals(op)) {
            broadcastState.remove(sourceTableName);
            configMap.remove(sourceTableName);
        } else {
            broadcastState.put(sourceTableName, value);
            configMap.put(sourceTableName,value);
        }
    }
    //JSONObject ：{"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"/abc","id":1,"type":"update"}
    //sinkColumns ：id,tm_name
    public static void deleteColNotNeed(JSONObject dataJsonObj, String sinkColumns) {
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        List<String> colList = Arrays.asList(sinkColumns.split(","));
        Iterator<Map.Entry<String, Object>> iterator = entrySet.iterator();
        for (;iterator.hasNext();){
            Map.Entry<String, Object> entry = iterator.next();
            if(!colList.contains(entry.getKey())){
                iterator.remove();
            }
        }
    }

}
